package com.xian.zk.common.config;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.xian.zk.common.properties.ZookeeperProperties;
import com.xian.zk.listener.ServerNodeChildrenDateListener;
import com.xian.zk.model.Instance;
import com.xian.zk.model.ServerNode;
import com.xian.zk.util.IpAddress;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @program: zoo-task
 * @description:
 * @author: liru.xian
 * @create: 2022-01-14 12:33
 **/
@Slf4j
@Configuration
public class ZookeeperConfig {

    @Autowired
    private ZookeeperProperties zookeeperProperties;

    private CuratorFramework client;

    @PostConstruct
    public void init() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookeeperProperties.getBaseSleepTimeMs(), zookeeperProperties.getMaxRetries());
        client = CuratorFrameworkFactory.builder().connectString(zookeeperProperties.getAddress()).retryPolicy(retryPolicy)
                .sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMs()).connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMs()).build();
        client.start();
        register();
    }

    @Deprecated
    public void stop() {
        client.close();
    }

    @Bean
    public CuratorFramework getClient() {
        return client;
    }

    /**
     * 注册节点
     */
    public void register() {
        try {
            ZooKeeper zk = client.getZookeeperClient().getZooKeeper();
            String registryPath = "/"+zookeeperProperties.getRegistryPath();
            if (zk.exists(registryPath, false) == null) {
                zk.create(registryPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                log.info("create registry node:{}", registryPath);
            }
            //创建服务节点（持久节点）
            String servicePath = registryPath + "/" + zookeeperProperties.getServerName();
            if (zk.exists(servicePath, false) == null) {
                zk.create(servicePath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                log.info("create service node:{}", servicePath);
            }
            //创建data 存储节点
            String dataPath = servicePath + "/data-";
            ObjectMapper mapper = new ObjectMapper();
            ServerNode serverNode = zookeeperProperties.getNode();
            serverNode.setServiceName(zookeeperProperties.getServerName());
            Instance instance = serverNode.getHosts().get(0);
            instance.setIp(IpAddress.get());
            instance.setServiceName(zookeeperProperties.getServerName());
            instance.setInstanceId(zookeeperProperties.getInstanceId());
            serverNode.getHosts().clear();
            serverNode.getHosts().add(instance);
            String dataJson = mapper.writeValueAsString(serverNode);
            String dataNode = zk.create(dataPath, dataJson.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            log.info("create data node:{} => {}", dataNode, dataJson);
            byte[] bytes = client.getData().forPath(dataNode);
            log.info("dataNode data {}",new String(bytes));

            PathChildrenCache childrenCache = new PathChildrenCache(client, servicePath, true);
            childrenCache.getListenable().addListener(new ServerNodeChildrenDateListener(client,zookeeperProperties));
            log.info("Register zk watcher {} path successfully!",servicePath);
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        } catch (Exception e) {
            log.error("create node failure", e);
        }
    }

    public List<String> getChildren(String path) {
        List<String> childrenList = new ArrayList<>();
        try {
            childrenList = client.getChildren().forPath(path);
        } catch (Exception e) {
            log.error("获取子节点出错", e);
        }
        return childrenList;
    }

    public int getChildrenCount(String path) {
        return getChildren(path).size();
    }

    public List<String> getInstances() {
        return getChildren(zookeeperProperties.getRegistryPath());
    }

    public int getInstancesCount() {
        return getInstances().size();
    }
}
